#include "mixp_worker.h"

#include <signal.h>
#include <unistd.h>

typedef struct req_s
{
    int fd;
    char buf[1024];
} req_t;

static mixp_worker_t *g_worker = NULL;
static void on_sig_quit(int);
static void on_message(int, void *);
static void on_error(int, void *);

static void do_echo(void *);

mixp_worker_t *mixp_worker_new(mixp_worker_info_t *info)
{
    mixp_worker_t *worker = MALLOC(mixp_worker_t);

    info->pid = getpid();
    worker->info = info;
    worker->pid = info->pid;
    worker->ppid = getppid();

    worker->pool = mixp_threadpool_new(info->module->n_threads, -1);
    if (!worker->pool)
    {
        loge("create threadpool failed!");
        goto err;
    }

    worker->evbase = gevent_base_create();
    if (!worker->evbase)
    {
        loge("gevent_base_create failed!");
        goto err;
    }

    worker->ev = gevent_create(info->fd, on_message, NULL, on_error, (void *)worker);
    if (-1 == gevent_add(worker->evbase, worker->ev))
    {
        loge("event_add failed!");
        goto err;
    }

    worker->req = MALLOC(mixp_packet_t);

    g_worker = worker;

    return worker;

err:
    FREE(worker->ev);
    FREE(worker->evbase);
    FREE(worker->pool);
    FREE(worker);

    return NULL;
}

void mixp_worker_free(mixp_worker_t *worker)
{
    mixp_threadpool_delete(worker->pool, GRACEFULLY);

    gevent_del(worker->evbase, worker->ev);
    gevent_destroy(worker->ev);
    gevent_base_destroy(worker->evbase);
    mixp_module_free(worker->info->module);
    FREE(worker->info);
    FREE(worker->req);
    FREE(worker);
}

void mixp_worker_loop(mixp_worker_t *worker)
{
    signal(SIGUSR1, on_sig_quit);
    // logi("begin module: %s, n_threads: %d", worker->info->module->name, worker->info->module->n_threads);

    uint8_t n_threads = worker->info->module->n_threads;
    int package_len = sizeof(mixp_packet_t) + sizeof(uint8_t);
    mixp_packet_t *packet_register = (mixp_packet_t *)malloc(package_len);
    packet_register->header.type = PACKET_WORKER_REGISTER;
    packet_register->header.payload_len = sizeof(uint8_t);
    memcpy(&packet_register->payload, &n_threads, sizeof(uint8_t));

    int n = skt_send(worker->info->fd, packet_register, package_len);
    if (n == -1)
    {
        printf("skt_send failed!\n");
    }

    FREE(packet_register);

    gevent_base_loop(worker->evbase);
}

static void on_sig_quit(int signum)
{
    logi("received quit signal.");
    gevent_base_loop_stop(g_worker->evbase);
}

static void on_message(int fd, void *arg)
{
    mixp_worker_t *worker = (mixp_worker_t *)arg;
    req_t *req = MALLOC(req_t);

    req->fd = fd;
    int len = skt_recv(fd, req->buf, sizeof(req->buf));
    if (len == 0)
    {
        loge("[ on_message ] master err.");
    }
    else
    {
        int ret = mixp_threadpool_add_task(worker->pool, do_echo, req, 0);
        if (ret)
            loge("[ on_message ] %s, add ret: %d.", req->buf, ret);
    }
}
static void on_error(int fd, void *arg)
{
    loge("[ on_error ] error: %d", errno);
}

static void do_echo(void *arg)
{
    req_t *req = (req_t *)arg;
    logi("[ do_echo ] %s.", req->buf);
    FREE(req);
}